/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.sshd.sftp.client;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.LinkOption;
import java.nio.file.Path;
import java.nio.file.attribute.AclEntry;
import java.nio.file.attribute.AclEntryFlag;
import java.nio.file.attribute.AclEntryPermission;
import java.nio.file.attribute.AclEntryType;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import org.apache.sshd.client.session.ClientSession;
import org.apache.sshd.common.util.GenericUtils;
import org.apache.sshd.common.util.MapEntryUtils;
import org.apache.sshd.common.util.MapEntryUtils.NavigableMapBuilder;
import org.apache.sshd.common.util.OsUtils;
import org.apache.sshd.common.util.io.IoUtils;
import org.apache.sshd.server.channel.ChannelSession;
import org.apache.sshd.server.command.Command;
import org.apache.sshd.server.session.ServerSession;
import org.apache.sshd.server.subsystem.SubsystemFactory;
import org.apache.sshd.sftp.SftpModuleProperties;
import org.apache.sshd.sftp.client.SftpClient.Attributes;
import org.apache.sshd.sftp.client.SftpClient.CloseableHandle;
import org.apache.sshd.sftp.client.SftpClient.CopyMode;
import org.apache.sshd.sftp.client.SftpClient.DirEntry;
import org.apache.sshd.sftp.client.SftpClient.OpenMode;
import org.apache.sshd.sftp.common.SftpConstants;
import org.apache.sshd.sftp.common.SftpException;
import org.apache.sshd.sftp.common.SftpHelper;
import org.apache.sshd.sftp.server.AbstractSftpEventListenerAdapter;
import org.apache.sshd.sftp.server.DefaultGroupPrincipal;
import org.apache.sshd.sftp.server.SftpEventListener;
import org.apache.sshd.sftp.server.SftpSubsystem;
import org.apache.sshd.sftp.server.SftpSubsystemEnvironment;
import org.apache.sshd.sftp.server.SftpSubsystemFactory;
import org.apache.sshd.util.test.CommonTestSupportUtils;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.MethodOrderer.MethodName;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * @author <a href="mailto:dev@mina.apache.org">Apache MINA SSHD Project</a>
 */
@TestMethodOrder(MethodName.class) // see https://github.com/junit-team/junit/wiki/Parameterized-tests
public class SftpVersionsTest extends AbstractSftpClientTestSupport {
    private static final List<Integer> VERSIONS = Collections.unmodifiableList(
            IntStream.rangeClosed(SftpSubsystemEnvironment.LOWER_SFTP_IMPL, SftpSubsystemEnvironment.HIGHER_SFTP_IMPL)
                    .boxed()
                    .collect(Collectors.toList()));

    private int testVersion;

    public void initSftpVersionsTest(int version) throws Exception {
        testVersion = version;
        setUp();
    }

    public static Collection<Object[]> parameters() {
        return parameterize(VERSIONS);
    }

    void setUp() throws Exception {
        setupServer();

        Map<String, Object> props = sshd.getProperties();
        Object forced = props.remove(SftpModuleProperties.SFTP_VERSION.getName());
        if (forced != null) {
            outputDebugMessage("Removed forced version=%s", forced);
        }
    }

    public final int getTestedVersion() {
        return testVersion;
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}") // See SSHD-749
    public void sftpOpenFlags(int version) throws Exception {
        initSftpVersionsTest(version);
        Path targetPath = detectTargetFolder();
        Path lclSftp
                = CommonTestSupportUtils.resolve(targetPath, SftpConstants.SFTP_SUBSYSTEM_NAME, getClass().getSimpleName());
        Path lclParent = assertHierarchyTargetFolderExists(lclSftp);
        Path lclFile = lclParent.resolve(getCurrentTestName() + "-" + getTestedVersion() + ".txt");
        Files.deleteIfExists(lclFile);

        Path parentPath = targetPath.getParent();
        String remotePath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, lclFile);
        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            try (OutputStream out = sftp.write(remotePath, OpenMode.Create, OpenMode.Write)) {
                out.write(getCurrentTestName().getBytes(StandardCharsets.UTF_8));
            }
            assertTrue(Files.exists(lclFile), "File should exist on disk: " + lclFile);
            sftp.remove(remotePath);
        }
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}")
    public void sftpCreateNew(int version) throws Exception {
        initSftpVersionsTest(version);
        Path targetPath = detectTargetFolder();
        Path lclSftp = CommonTestSupportUtils.resolve(targetPath, SftpConstants.SFTP_SUBSYSTEM_NAME,
                getClass().getSimpleName());
        Path lclParent = assertHierarchyTargetFolderExists(lclSftp);
        Path lclFile = lclParent.resolve(getCurrentTestName() + "-" + getTestedVersion() + ".txt");
        Files.write(lclFile, Collections.singleton("existing"));

        Path parentPath = targetPath.getParent();
        String remotePath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, lclFile);
        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            SftpException ex = assertThrows(SftpException.class, () -> {
                try (OutputStream out = sftp.write(remotePath, OpenMode.Create, OpenMode.Write, OpenMode.Exclusive)) {
                    out.write(getCurrentTestName().getBytes(StandardCharsets.UTF_8));
                }
            });
            assertEquals(SftpConstants.SSH_FX_FILE_ALREADY_EXISTS, ex.getStatus());
        }
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}")
    public void sftpRenameNoReplace(int version) throws Exception {
        initSftpVersionsTest(version);
        Path targetPath = detectTargetFolder();
        Path lclSftp = CommonTestSupportUtils.resolve(targetPath, SftpConstants.SFTP_SUBSYSTEM_NAME,
                getClass().getSimpleName());
        Path lclParent = assertHierarchyTargetFolderExists(lclSftp);
        Path aFile = lclParent.resolve(getCurrentTestName() + "-" + getTestedVersion() + "-a.txt");
        Files.write(aFile, Collections.singleton("a"));
        Path bFile = lclParent.resolve(getCurrentTestName() + "-" + getTestedVersion() + "-b.txt");
        Files.write(bFile, Collections.singleton("b"));

        Path parentPath = targetPath.getParent();
        String aPath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, aFile);
        String bPath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, bFile);
        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            SftpException ex = assertThrows(SftpException.class, () -> sftp.rename(aPath, bPath));
            assertEquals(SftpConstants.SSH_FX_FILE_ALREADY_EXISTS, ex.getStatus());
            if (getTestedVersion() >= SftpConstants.SFTP_V5) {
                // For CopyMode.Atomic we use StandardCopyOptions.ATOMIC_MOVE. It is implementation defined whether an
                // atomic move overwrites an already existing file. See javadoc of Files.move().
                sftp.rename(aPath, bPath, CopyMode.Overwrite);
                assertTrue(Files.notExists(aFile));
                try (InputStream in = sftp.read(bPath)) {
                    List<String> lines = IoUtils.readAllLines(in);
                    assertEquals(1, lines.size());
                    assertEquals("a", lines.get(0));
                }
            } else {
                assertThrows(UnsupportedOperationException.class, () -> sftp.rename(aPath, bPath, CopyMode.Atomic));
                assertThrows(UnsupportedOperationException.class, () -> sftp.rename(aPath, bPath, CopyMode.Overwrite));
            }
        }
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}")
    public void sftpVersionSelector(int version) throws Exception {
        initSftpVersionsTest(version);
        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            assertEquals(getTestedVersion(), sftp.getVersion(), "Mismatched negotiated version");
        }
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}") // see SSHD-572
    public void sftpFileTimesUpdate(int version) throws Exception {
        initSftpVersionsTest(version);
        Path targetPath = detectTargetFolder();
        Path lclSftp
                = CommonTestSupportUtils.resolve(targetPath, SftpConstants.SFTP_SUBSYSTEM_NAME, getClass().getSimpleName());
        Path lclFile
                = assertHierarchyTargetFolderExists(lclSftp).resolve(getCurrentTestName() + "-" + getTestedVersion() + ".txt");
        Files.write(lclFile, getClass().getName().getBytes(StandardCharsets.UTF_8));
        Path parentPath = targetPath.getParent();
        String remotePath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, lclFile);
        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            Attributes attrs = sftp.lstat(remotePath);
            long expectedSeconds = TimeUnit.SECONDS.convert(System.currentTimeMillis() - TimeUnit.HOURS.toMillis(1L),
                    TimeUnit.MILLISECONDS);
            attrs.getFlags().clear();
            attrs.modifyTime(expectedSeconds);
            sftp.setStat(remotePath, attrs);

            attrs = sftp.lstat(remotePath);
            long actualSeconds = attrs.getModifyTime().to(TimeUnit.SECONDS);
            // The NTFS file system delays updates to the last access time for a file by up to 1 hour after the last
            // access
            if (expectedSeconds != actualSeconds) {
                System.err.append("Mismatched last modified time for ").append(lclFile.toString())
                        .append(" - expected=").append(String.valueOf(expectedSeconds))
                        .append('[').append(new Date(TimeUnit.SECONDS.toMillis(expectedSeconds)).toString()).append(']')
                        .append(", actual=").append(String.valueOf(actualSeconds))
                        .append('[').append(new Date(TimeUnit.SECONDS.toMillis(actualSeconds)).toString()).append(']')
                        .println();
            }
        }
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}") // see SSHD-573
    public void sftpFileTypeAndPermissionsUpdate(int version) throws Exception {
        initSftpVersionsTest(version);
        Path targetPath = detectTargetFolder();
        Path lclSftp
                = CommonTestSupportUtils.resolve(targetPath, SftpConstants.SFTP_SUBSYSTEM_NAME, getClass().getSimpleName());
        Path subFolder = Files.createDirectories(lclSftp.resolve("sub-folder"));
        String subFolderName = subFolder.getFileName().toString();
        Path lclFile
                = assertHierarchyTargetFolderExists(lclSftp).resolve(getCurrentTestName() + "-" + getTestedVersion() + ".txt");
        String lclFileName = lclFile.getFileName().toString();
        Files.write(lclFile, getClass().getName().getBytes(StandardCharsets.UTF_8));

        Path parentPath = targetPath.getParent();
        String remotePath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, lclSftp);
        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            for (DirEntry entry : sftp.readDir(remotePath)) {
                String fileName = entry.getFilename();
                if (".".equals(fileName) || "..".equals(fileName)) {
                    continue;
                }

                Attributes attrs = validateSftpFileTypeAndPermissions(fileName, getTestedVersion(), entry.getAttributes());
                if (subFolderName.equals(fileName)) {
                    assertEquals(SftpConstants.SSH_FILEXFER_TYPE_DIRECTORY, attrs.getType(), "Mismatched sub-folder type");
                    assertTrue(attrs.isDirectory(), "Sub-folder not marked as directory");
                } else if (lclFileName.equals(fileName)) {
                    assertEquals(SftpConstants.SSH_FILEXFER_TYPE_REGULAR, attrs.getType(), "Mismatched sub-file type");
                    assertTrue(attrs.isRegularFile(), "Sub-folder not marked as directory");
                }
            }
        }
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}") // see SSHD-574
    public void sftpACLEncodeDecode(int version) throws Exception {
        initSftpVersionsTest(version);
        AclEntryType[] types = AclEntryType.values();
        final List<AclEntry> aclExpected = new ArrayList<>(types.length);
        for (AclEntryType t : types) {
            aclExpected.add(AclEntry.newBuilder()
                    .setType(t)
                    .setFlags(EnumSet.allOf(AclEntryFlag.class))
                    .setPermissions(EnumSet.allOf(AclEntryPermission.class))
                    .setPrincipal(new DefaultGroupPrincipal(getCurrentTestName() + "@" + getClass().getPackage().getName()))
                    .build());
        }

        AtomicInteger numInvocations = new AtomicInteger(0);
        SftpSubsystemFactory factory = new SftpSubsystemFactory() {
            @Override
            public Command createSubsystem(ChannelSession channel) throws IOException {
                SftpSubsystem subsystem = new SftpSubsystem(channel, this) {
                    @Override
                    protected NavigableMap<String, Object> resolveFileAttributes(
                            Path file, int flags, boolean neverFollowSymLinks, LinkOption... options)
                            throws IOException {
                        NavigableMap<String, Object> attrs
                                = super.resolveFileAttributes(file, flags, neverFollowSymLinks, options);
                        if (MapEntryUtils.isEmpty(attrs)) {
                            attrs = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
                        }

                        @SuppressWarnings("unchecked")
                        List<AclEntry> aclActual = (List<AclEntry>) attrs.put(IoUtils.ACL_VIEW_ATTR, aclExpected);
                        if (aclActual != null) {
                            log.info("resolveFileAttributes(" + file + ") replaced ACL: " + aclActual);
                        }
                        return attrs;
                    }

                    @Override
                    protected void setFileAccessControl(Path file, List<AclEntry> aclActual, LinkOption... options)
                            throws IOException {
                        if (aclActual != null) {
                            assertListEquals("Mismatched ACL set for file=" + file, aclExpected, aclActual);
                            numInvocations.incrementAndGet();
                        }
                    }
                };
                Collection<? extends SftpEventListener> listeners = getRegisteredListeners();
                if (GenericUtils.size(listeners) > 0) {
                    for (SftpEventListener l : listeners) {
                        subsystem.addSftpEventListener(l);
                    }
                }

                return subsystem;
            }
        };

        factory.addSftpEventListener(new AbstractSftpEventListenerAdapter() {
            @Override
            public void modifyingAttributes(ServerSession session, Path path, Map<String, ?> attrs) {
                @SuppressWarnings("unchecked")
                List<AclEntry> aclActual
                        = MapEntryUtils.isEmpty(attrs) ? null : (List<AclEntry>) attrs.get(IoUtils.ACL_VIEW_ATTR);
                if (getTestedVersion() > SftpConstants.SFTP_V3) {
                    assertListEquals("Mismatched modifying ACL for file=" + path, aclExpected, aclActual);
                } else {
                    assertNull(aclActual, "Unexpected modifying ACL for file=" + path);
                }
            }

            @Override
            public void modifiedAttributes(
                    ServerSession session, Path path, Map<String, ?> attrs, Throwable thrown) {
                @SuppressWarnings("unchecked")
                List<AclEntry> aclActual
                        = MapEntryUtils.isEmpty(attrs) ? null : (List<AclEntry>) attrs.get(IoUtils.ACL_VIEW_ATTR);
                if (getTestedVersion() > SftpConstants.SFTP_V3) {
                    assertListEquals("Mismatched modified ACL for file=" + path, aclExpected, aclActual);
                } else {
                    assertNull(aclActual, "Unexpected modified ACL for file=" + path);
                }
            }
        });

        Path targetPath = detectTargetFolder();
        Path lclSftp = CommonTestSupportUtils.resolve(
                targetPath, SftpConstants.SFTP_SUBSYSTEM_NAME, getClass().getSimpleName());
        Files.createDirectories(lclSftp.resolve("sub-folder"));
        Path lclFile = assertHierarchyTargetFolderExists(lclSftp)
                .resolve(getCurrentTestName() + "-" + getTestedVersion() + ".txt");
        Files.write(lclFile, getClass().getName().getBytes(StandardCharsets.UTF_8));

        Path parentPath = targetPath.getParent();
        String remotePath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, lclSftp);
        int numInvoked = 0;

        List<? extends SubsystemFactory> factories = sshd.getSubsystemFactories();
        sshd.setSubsystemFactories(Collections.singletonList(factory));
        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            for (DirEntry entry : sftp.readDir(remotePath)) {
                String fileName = entry.getFilename();
                if (".".equals(fileName) || "..".equals(fileName)) {
                    continue;
                }

                Attributes attrs = validateSftpFileTypeAndPermissions(fileName, getTestedVersion(), entry.getAttributes());
                List<AclEntry> aclActual = attrs.getAcl();
                if (getTestedVersion() == SftpConstants.SFTP_V3) {
                    assertNull(aclActual, "Unexpected ACL for entry=" + fileName);
                } else {
                    assertListEquals("Mismatched ACL for entry=" + fileName, aclExpected, aclActual);
                }

                attrs.getFlags().clear();
                attrs.setAcl(aclExpected);
                sftp.setStat(remotePath + "/" + fileName, attrs);
                if (getTestedVersion() > SftpConstants.SFTP_V3) {
                    numInvoked++;
                }
            }
        } finally {
            sshd.setSubsystemFactories(factories);
        }

        assertEquals(numInvoked, numInvocations.get(), "Mismatched invocations count");
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}") // see SSHD-575
    public void sftpExtensionsEncodeDecode(int version) throws Exception {
        initSftpVersionsTest(version);
        Class<?> anchor = getClass();
        Map<String, String> expExtensions = NavigableMapBuilder.<String, String> builder(String.CASE_INSENSITIVE_ORDER)
                .put("class", anchor.getSimpleName())
                .put("package", anchor.getPackage().getName())
                .put("method", getCurrentTestName())
                .build();

        final AtomicInteger numInvocations = new AtomicInteger(0);
        SftpSubsystemFactory factory = new SftpSubsystemFactory() {
            @Override
            public Command createSubsystem(ChannelSession channel) throws IOException {
                SftpSubsystem subsystem = new SftpSubsystem(channel, this) {
                    @Override
                    protected NavigableMap<String, Object> resolveFileAttributes(
                            Path file, int flags, boolean neverFollowLinks, LinkOption... options)
                            throws IOException {
                        NavigableMap<String, Object> attrs
                                = super.resolveFileAttributes(file, flags, neverFollowLinks, options);
                        if (MapEntryUtils.isEmpty(attrs)) {
                            attrs = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
                        }

                        @SuppressWarnings("unchecked")
                        Map<String, String> actExtensions
                                = (Map<String, String>) attrs.put(IoUtils.EXTENDED_VIEW_ATTR, expExtensions);
                        if (actExtensions != null) {
                            log.info("resolveFileAttributes(" + file + ") replaced extensions: " + actExtensions);
                        }
                        return attrs;
                    }

                    @Override
                    protected void setFileExtensions(Path file, Map<String, byte[]> extensions, LinkOption... options)
                            throws IOException {
                        assertExtensionsMapEquals("setFileExtensions(" + file + ")", expExtensions, extensions);
                        numInvocations.incrementAndGet();

                        int currentVersion = getTestedVersion();
                        try {
                            super.setFileExtensions(file, extensions, options);
                            assertFalse(currentVersion >= SftpConstants.SFTP_V6,
                                    "Expected exception not generated for version=" + currentVersion);
                        } catch (UnsupportedOperationException e) {
                            assertTrue(currentVersion >= SftpConstants.SFTP_V6,
                                    "Unexpected exception for version=" + currentVersion);
                        }
                    }
                };
                Collection<? extends SftpEventListener> listeners = getRegisteredListeners();
                if (GenericUtils.size(listeners) > 0) {
                    for (SftpEventListener l : listeners) {
                        subsystem.addSftpEventListener(l);
                    }
                }

                return subsystem;
            }
        };

        factory.addSftpEventListener(new AbstractSftpEventListenerAdapter() {
            @Override
            public void modifyingAttributes(ServerSession session, Path path, Map<String, ?> attrs) {
                @SuppressWarnings("unchecked")
                Map<String, byte[]> actExtensions
                        = MapEntryUtils.isEmpty(attrs) ? null : (Map<String, byte[]>) attrs.get(IoUtils.EXTENDED_VIEW_ATTR);
                assertExtensionsMapEquals("modifying(" + path + ")", expExtensions, actExtensions);
            }

            @Override
            public void modifiedAttributes(ServerSession session, Path path, Map<String, ?> attrs, Throwable thrown) {
                @SuppressWarnings("unchecked")
                Map<String, byte[]> actExtensions
                        = MapEntryUtils.isEmpty(attrs) ? null : (Map<String, byte[]>) attrs.get(IoUtils.EXTENDED_VIEW_ATTR);
                assertExtensionsMapEquals("modified(" + path + ")", expExtensions, actExtensions);
            }
        });

        Path targetPath = detectTargetFolder();
        Path lclSftp
                = CommonTestSupportUtils.resolve(targetPath, SftpConstants.SFTP_SUBSYSTEM_NAME, getClass().getSimpleName());
        Files.createDirectories(lclSftp.resolve("sub-folder"));
        Path lclFile
                = assertHierarchyTargetFolderExists(lclSftp).resolve(getCurrentTestName() + "-" + getTestedVersion() + ".txt");
        Files.write(lclFile, getClass().getName().getBytes(StandardCharsets.UTF_8));

        Path parentPath = targetPath.getParent();
        String remotePath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, lclSftp);
        int numInvoked = 0;

        List<? extends SubsystemFactory> factories = sshd.getSubsystemFactories();
        sshd.setSubsystemFactories(Collections.singletonList(factory));
        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            for (DirEntry entry : sftp.readDir(remotePath)) {
                String fileName = entry.getFilename();
                if (".".equals(fileName) || "..".equals(fileName)) {
                    continue;
                }

                Attributes attrs = validateSftpFileTypeAndPermissions(fileName, getTestedVersion(), entry.getAttributes());
                Map<String, byte[]> actExtensions = attrs.getExtensions();
                assertExtensionsMapEquals("dirEntry=" + fileName, expExtensions, actExtensions);
                attrs.getFlags().clear();
                attrs.setStringExtensions(expExtensions);
                sftp.setStat(remotePath + "/" + fileName, attrs);
                numInvoked++;
            }
        } finally {
            sshd.setSubsystemFactories(factories);
        }

        assertEquals(numInvoked, numInvocations.get(), "Mismatched invocations count");
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}") // see SSHD-623
    public void endOfListIndicator(int version) throws Exception {
        initSftpVersionsTest(version);
        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            AtomicReference<Boolean> eolIndicator = new AtomicReference<>();
            Path targetPath = detectTargetFolder();
            Path parentPath = targetPath.getParent();
            String remotePath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, targetPath);

            try (CloseableHandle handle = sftp.openDir(remotePath)) {
                List<DirEntry> entries = sftp.readDir(handle, eolIndicator);
                for (int index = 1; entries != null; entries = sftp.readDir(handle, eolIndicator), index++) {
                    Boolean value = eolIndicator.get();
                    if (version < SftpConstants.SFTP_V6) {
                        assertNull(value, "Unexpected indicator value at iteration #" + index);
                    } else {
                        assertNotNull(value, "No indicator returned at iteration #" + index);
                        if (value) {
                            break;
                        }
                    }
                    eolIndicator.set(null); // make sure starting fresh
                }

                Boolean value = eolIndicator.get();
                if (version < SftpConstants.SFTP_V6) {
                    assertNull(value, "Unexpected end-of-list indication received at end of entries");
                    assertNull(entries, "Unexpected no last entries indication");
                } else {
                    assertNotNull(value, "No end-of-list indication received at end of entries");
                    assertNotNull(entries, "No last received entries");
                    assertTrue(value, "Bad end-of-list value");
                }
            }
        }
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}")
    public void sftpRemoveFileWithSymlinkInPath(int version) throws Exception {
        Assumptions.assumeFalse(OsUtils.isWin32(), "Symlinks not reliably supported on Windows");

        initSftpVersionsTest(version);
        Path targetPath = detectTargetFolder();
        Path lclSftp = CommonTestSupportUtils.resolve(
                targetPath,
                SftpConstants.SFTP_SUBSYSTEM_NAME,
                getClass().getSimpleName());
        Path lclParent = assertHierarchyTargetFolderExists(lclSftp);
        String testFileName = "test-file.txt";

        Path actualDir = lclParent.resolve("actual-dir-" + getTestedVersion());
        CommonTestSupportUtils.deleteRecursive(actualDir);
        Files.createDirectories(actualDir);

        Path symlinkDir = lclParent.resolve("symlink-dir-" + getTestedVersion());
        Files.deleteIfExists(symlinkDir);
        Files.createSymbolicLink(symlinkDir, actualDir);

        Path fileViaSymlink = symlinkDir.resolve(testFileName);
        Files.write(fileViaSymlink, "test content".getBytes(StandardCharsets.UTF_8));

        assertTrue(Files.exists(fileViaSymlink), "File should exist via symlink path");
        assertTrue(Files.exists(actualDir.resolve(testFileName)), "File should exist in actual directory");

        Path parentPath = targetPath.getParent();
        String remotePath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, fileViaSymlink);

        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            sftp.remove(remotePath);

            assertFalse(Files.exists(fileViaSymlink), "File should be deleted via symlink path");
            assertFalse(Files.exists(actualDir.resolve(testFileName)), "File should be deleted from actual directory");
        } finally {
            Files.deleteIfExists(symlinkDir);
            CommonTestSupportUtils.deleteRecursive(actualDir);
        }
    }

    @MethodSource("parameters")
    @ParameterizedTest(name = "version={0}")
    public void sftpFStatWithSymlinkInPath(int version) throws Exception {
        Assumptions.assumeFalse(OsUtils.isWin32(), "Symlinks not reliably supported on Windows");

        initSftpVersionsTest(version);
        Path targetPath = detectTargetFolder();
        Path lclSftp = CommonTestSupportUtils.resolve(targetPath,
                SftpConstants.SFTP_SUBSYSTEM_NAME, getClass().getSimpleName());
        Path lclParent = assertHierarchyTargetFolderExists(lclSftp);

        Path actualDir = lclParent.resolve("actual-fstat-" + getTestedVersion());
        CommonTestSupportUtils.deleteRecursive(actualDir);
        Files.createDirectories(actualDir);

        Path symlinkDir = lclParent.resolve("symlink-fstat-" + getTestedVersion());
        Files.deleteIfExists(symlinkDir);
        Files.createSymbolicLink(symlinkDir, actualDir);

        Path testFile = symlinkDir.resolve("test.txt");
        String content = getCurrentTestName();
        Files.write(testFile, content.getBytes(StandardCharsets.UTF_8));

        Path parentPath = targetPath.getParent();
        String remotePath = CommonTestSupportUtils.resolveRelativeRemotePath(parentPath, testFile);

        try (ClientSession session = createAuthenticatedClientSession();
             SftpClient sftp = createSftpClient(session, getTestedVersion())) {
            try (SftpClient.CloseableHandle handle = sftp.open(remotePath, SftpClient.OpenMode.Read)) {
                Attributes attrs = sftp.stat(handle);

                assertNotNull(attrs, "Attributes should be retrieved");
                assertEquals(content.length(), attrs.getSize(), "File size mismatch");
                assertTrue(attrs.isRegularFile(), "Should be a regular file");
            }
        } finally {
            Files.deleteIfExists(testFile);
            Files.deleteIfExists(symlinkDir);
            CommonTestSupportUtils.deleteRecursive(actualDir);
        }
    }

    @Override
    public String toString() {
        return getClass().getSimpleName() + "[" + getTestedVersion() + "]";
    }

    public static void assertExtensionsMapEquals(String message, Map<String, String> expected, Map<String, byte[]> actual) {
        assertMapEquals(message, expected, SftpHelper.toStringExtensions(actual));
    }

    private static Attributes validateSftpFileTypeAndPermissions(String fileName, int version, Attributes attrs) {
        int actualPerms = attrs.getPermissions();
        if (version == SftpConstants.SFTP_V3) {
            int expected = SftpHelper.permissionsToFileType(actualPerms);
            assertEquals(expected, attrs.getType(), fileName + ": Mismatched file type");
        } else {
            int expected = SftpHelper.fileTypeToPermission(attrs.getType());
            assertEquals((actualPerms & expected), expected,
                    fileName + ": Missing permision=0x" + Integer.toHexString(expected) + " in 0x"
                                                             + Integer.toHexString(actualPerms));
        }

        return attrs;
    }
}
